package com.zhaokuner.springboot.zk;

import com.zhaokuner.springboot.refresh.RefreshScopeLaunch;
import lombok.Data;
import lombok.extern.log4j.Log4j2;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import javax.annotation.PreDestroy;

/**
 * <p>
 * Description:zk 事件 通知
 * <p>
 * Copyright: 2018-现在 北京掌通未来科技有限公司
 * <p>
 *
 * @author 赵希坤 zhaoxikun
 * @date 2020/7/25 14:36
 */
@Log4j2
@Data
public class CuratorClientRefreshLaunch implements ApplicationRunner {
    /**
     * zk说明<p>
     * zookeeper.extendedTypesEnabled ：(仅 Java 系统 property：zookeeper.extendedTypesEnabled)
     * **3.5.4 的新功能：**定义为“ true”以启用扩展的 features(例如创建 TTL 节点)。
     * 默认情况下禁用它们。重要信息：由于内部限制，启用的服务器 ID 必须小于 255。
     * <p>
     * zookeeper.emulate353TTLNodes ：(仅 Java 系统 property：zookeeper.emulate353TTLNodes)
     * 3.5.4 的新功能：由于 ZOOKEEPER-2901，3.5.4/3.6.0 不支持在 version 3.5.3 中创建的 TTL 节点。
     * 但是，通过 zookeeper.emulate353TTLNodes 系统 property 提供了解决方法。
     * 如果您在 ZooKeeper 3.5.3 中使用了 TTL 节点，并且除了zookeeper.extendedTypesEnabled之外，还需要将兼容性zookeeper.emulate353TTLNodes设置为“ true”。
     * 注意：由于该错误，服务器 ID 必须小于或等于 127。此外，最大支持 TTL value 是 1099511627775，小于 3.5.3 中允许的值(1152921504606846975)
     */
    private CuratorFramework client = null;
    private CuratorCache cacheDefault = null;
    private CuratorCache cacheProfile = null;

    @Value("${spring.cloud.config.serverName}")
    private String configServerName = "spring_cloud_config_server";
    @Value("${spring.cloud.config.label}")
    private String label;
    @Value("${spring.profiles.active}")
    private String profile;
    @Value("${spring.cloud.config.zkConnectString}")
    private String zkConnectString;
    @Value("${spring.application.name}")
    private String applicationName;
    @Autowired
    private RefreshScopeLaunch refreshScopeLaunch;
    /**
     * 5分钟
     */
    private long lastTime = 1000 * 60 * 5;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        init();
        cacheDefault = addCuratorCacheWatcher(applicationName);
        if (null != profile && !profile.isEmpty()) {
            cacheProfile = addCuratorCacheWatcher(applicationName + "-" + profile);
        }
        log.info("~~~~~~~~~~~~~~~~~~~~CuratorClientRefreshLaunch config over");
    }

    public void init() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString(zkConnectString)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        log.info("~~~~~~~~~~~~~~~~~~~~CuratorClientRefreshLaunch start success");
    }

    private CuratorCache addCuratorCacheWatcher(String fileName) {
        String path = "/" + configServerName + "/" + label + "/" + fileName;
        CuratorCache cache = CuratorCache.build(client, path);
        // there are several ways to set a listener on a CuratorCache. You can watch for individual events
        // or for all events. Here, we'll use the builder to log individual cache actions
        CuratorCacheListener listener = CuratorCacheListener.builder()
                .forCreates(node -> {
                    if (null == node.getData()) {
                        log.info(String.format("~~~~~~~~~~~~~~~~~~~~Node creates. [%s] [%s]", node.getPath(), "null"));
                        return;
                    }
                    //应用启动后，会立即触发一次，这里使用时间判断，git push 时间在5分钟内的create事件，进行刷新
                    String value = new String(node.getData());
                    int index = value.indexOf("-");
                    if (index == -1) {
                        //格式不对
                        log.info(String.format("~~~~~~~~~~~~~~~~~~~~Node creates.datetime is error. [%s] [%s]", node.getPath(), value));
                        return;
                    }
                    long time = 0;
                    try {
                        time = Long.parseLong(value.substring(0, index));
                    } catch (Exception e) {
                        //格式不对
                        log.info(String.format("~~~~~~~~~~~~~~~~~~~~Node creates.datetime is error. [%s] [%s]", node.getPath(), value));
                        return;
                    }
                    boolean isNotRefresh = System.currentTimeMillis() - time > lastTime;
                    log.info(String.format("~~~~~~~~~~~~~~~~~~~~Node creates. refresh [%s] [%s] [%s]", !isNotRefresh, node.getPath(), value));
                    if (isNotRefresh) {
                        return;
                    }
                    refreshScopeLaunch.refresh(value);
                })
                .forChanges((oldNode, node) -> {
                    byte[] oldByte = oldNode.getData();
                    if (null == oldByte) {
                        oldByte = new byte[0];
                    }
                    String oldValue = new String(oldByte);
                    byte[] newByte = node.getData();
                    if (null == newByte) {
                        newByte = new byte[0];
                    }
                    String newValue = new String(newByte);
                    log.info(String.format("~~~~~~~~~~~~~~~~~~~~Node changed. [%s] [%s] --> [%s]", oldNode.getPath(), oldValue, newValue));
                    refreshScopeLaunch.refresh(newValue);
                })
                .forInitialized(() -> log.info("~~~~~~~~~~~~~~~~~~~~Cache initialized " + path))
                .build();

        // register the listener
        cache.listenable().addListener(listener);

        // the cache must be started
        cache.start();
        log.info("~~~~~~~~~~~~~~~~~~~~CuratorClientRefreshLaunch addCuratorCacheWatcher success [" + path + "]");
        return cache;
    }


    @PreDestroy
    public void close() {
        closeCache();
        closeClient();
    }

    private void closeClient() {
        if (client != null) {
            client.close();
            log.info("~~~~~~~~~~~~~~~~~~~~CuratorClientRefreshLaunch close");
        }
    }

    private void closeCache() {
        if (cacheDefault != null) {
            cacheDefault.close();
            log.info("~~~~~~~~~~~~~~~~~~~~CuratorClientRefreshLaunch cache default close");
        }
        if (cacheProfile != null) {
            cacheProfile.close();
            log.info("~~~~~~~~~~~~~~~~~~~~CuratorClientRefreshLaunch cache {} close", profile);
        }
    }
}
